package kafka;

import net.sf.json.JSONObject;
import org.apache.log4j.Logger;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 从kafka 取数据清洗及预警
 *
 * USER: lintc 【lintiancong@zhuojianchina.com】
 * DATE: 2017-08-09 15:26
 */
public class IpWarn implements KafkaHandle {

    private static Logger LOG = Logger.getLogger(IpWarn.class);

    Map<String, Integer> ipMap = new HashMap<>();

    @Override
    public void handle(LogMap logMap, String topic, String time) {
        List<JSONObject> list = logMap.getList(time);
        for (JSONObject obj : list) {
            String ip = obj.optJSONObject("logs").optString("ip");
            getIp(time, ip, topic);
        }
    }

    public void getIp(String time, String ip, String topic) {
        String key = topic + "_" + time + "_" + ip;
        Integer count = ipMap.get(key);
        if (null == count) {
            count = 0;
        }
        ipMap.put(key, count + 1);
        if (count >= 20) {
            LOG.info(topic + " : " + ip + " : " + ipMap.get(key));
        }
    }
}
